use crate::router_service::ping_pong::PongRouterFactory;
use std::sync::Arc;
use std::time::{SystemTime};
use bytes::{BufMut, BytesMut};
use log::{error, info};
use tokio::io::{AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::{broadcast, mpsc};
use rpc::router::{RouterFactory};
use rpc::{time_to_millis, ChannelId, DefaultLocalSeqCreator, LocalSeqCreator, RpcContext, RpcEvent, RpcResult};
use rpc::bootstrap::{Bootstrap, BootstrapBuilder};
use rpc::codec::{RpcCodecGather};
use rpc::connection_manager::connection::RpcConnection;
use rpc::dto::read_packet_for_timeout;
use rpc::error::{CustomerErrorKind, RpcError};
use rpc::sender::RpcSender;
use rpc_derive::router;
use crate::connection_manager::server_connection::{ServerConnectionHandler};
use crate::connection_manager::ServerConnectionManager;

pub struct SocketServer {
    port: u32,
    bootstrap: Bootstrap,
    shutdown: broadcast::Sender<()>,
}

pub struct RpcServerBuilder {
    port: u32,
    bootstrap: Bootstrap
}

impl SocketServer {

    pub fn get_router(&self, path: &'static str) ->  Option<&Box<dyn RouterFactory + 'static>> {
        self.bootstrap.get_router(path)
    }

    pub async fn start_server(&self) -> RpcResult<()> {
        let server_connection_manager = ServerConnectionManager::new();
        let connection_manager_sender = rpc::connection_manager::start(server_connection_manager);

        let rpc_codec_map = self.bootstrap.rpc_codec_map.clone();
        let buff_size = self.bootstrap.buff_size;

        let address = format!("0.0.0.0:{}", self.port);
        let listener = TcpListener::bind(&address).await?;
        let mut connect_id = 0;

        let local_seq_creator = match &self.bootstrap.local_seq_creator {
            None => {
                let prefix = self.bootstrap.date_time_format();
                let default_creator = Box::new(DefaultLocalSeqCreator::new(prefix)) as Box<dyn LocalSeqCreator>;
                Arc::new(default_creator)
            }
            Some(local_seq_creator) => { local_seq_creator.clone() }
        };

        loop {
            let (mut stream, socket_addr) = listener.accept().await?;
            connect_id = connect_id + 1;
            info!("{} connected from {}", connect_id, socket_addr);

            let codec_type = self.read_codec_type_data(&mut stream, rpc_codec_map.clone()).await?;
            let (event_sender, event_receive) = mpsc::unbounded_channel::<Box<dyn RpcEvent>>();
            let sender = RpcSender::new(connection_manager_sender.clone(), event_sender);

            let sms = sender.clone();
            let router_clone = self.bootstrap.router.clone();
            let local_seq_creator_clone = local_seq_creator.clone();
            let rpc_codec_map_clone = rpc_codec_map.clone();
            let idle_listeners_clone = self.bootstrap.idle_listeners.clone();

            let close_receive = self.shutdown.subscribe();
            tokio::spawn(async move {
                let mut context = RpcContext::create(local_seq_creator_clone, rpc_codec_map_clone, sms.clone());
                context.set_current_codec_type(Some(codec_type)).unwrap();

                println!("idle listener length: {}", idle_listeners_clone.len());
                let mut connection = RpcConnection::new(ChannelId::new(connect_id), buff_size,
                                                        context.clone(), event_receive, idle_listeners_clone);

                let connection_handler = ServerConnectionHandler::new(router_clone.clone(),
                                                                      ChannelId::new(connect_id), sms);

                let run_result = connection.run(stream, connection_handler, ||{}, close_receive).await;
                if let Err(e) = run_result {
                    error!("Error running connection: {:?}", e);
                }
            });
        }
    }

    async fn read_codec_type_data(&self, stream: &mut TcpStream, codec_gather: RpcCodecGather) -> RpcResult<String> {
        let packet = read_packet_for_timeout(stream, 3000).await?;
        let codec_type = String::from_utf8(packet)?;
        if !codec_gather.has_codec_type(&codec_type) {
            let error = RpcError::CustomerError {error_kid:
            CustomerErrorKind::CodecTypeError, message: "codec_type错误".to_string()};
            return Err(error);
        }
        stream.write_all(&self.build_codec_response_data()).await?;
        Ok(codec_type)
    }

    fn build_codec_response_data(&self)-> Vec<u8> {
        let now = time_to_millis(SystemTime::now());
        let data = format!("success:{}", now);
        let data_bytes = data.as_bytes();
        let mut bytes_mut = BytesMut::with_capacity(1 + data_bytes.len());
        bytes_mut.put_u8(data.len() as u8);
        bytes_mut.put_slice(data_bytes);
        bytes_mut.to_vec()
    }
}

impl RpcServerBuilder {

    pub fn new(bootstrap_builder: BootstrapBuilder, port: u32) -> Self {
        let bb = bootstrap_builder.router("pong", router!(pong));
        let builder = RpcServerBuilder {
            bootstrap: bb.build(),
            port,
        };
        builder
    }

    pub fn build(self) -> SocketServer {
        let (close_send, close_receive) = broadcast::channel::<()>(1);
        SocketServer {
            port: self.port,
            bootstrap: self.bootstrap,
            shutdown: close_send
        }
    }
}
